package com.doubao.hardware.config;

import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * <p>MQTT客户端</p>
 * 使用方式：
 * <pre>
 * IotMqttClient.create()
 *              .broker(broker)
 *              .auth(username, password)
 *              .clientId(clientId)
 *              .cleanSession(true)
 *              .automaticReconnect(true)
 *              .subscribe(topic, 1)
 *              .callback(new MqttCallbackHandler())
 *              .connect();
 * </pre>
 */
public class IotMqttClient {

    private static final Logger LOG = LoggerFactory.getLogger(IotMqttClient.class);
    private static final String logTitle = "MQTT信息: ";

    private final List<SubTopic> subTopics = new ArrayList<>();
    private MqttClientPersistence persistence;

    private MqttClient client;
    private MqttConnectOptions options;
    private String broker;
    private String clientId;
    private String username;
    private String password;
    private boolean cleanSession = MqttConnectOptions.CLEAN_SESSION_DEFAULT;
    private boolean automaticReconnect = false;
    private BaseMqttCallback callback;
    private int defaultQos = 0;
    /**
     * 连接次数
     */
    private final AtomicInteger connectTimes = new AtomicInteger();

    private IotMqttClient() {
    }

    /**
     * 创建实例
     *
     * @return 返回IotMqttClient对象
     */
    public static IotMqttClient create() {
        return new IotMqttClient();
    }

    /**
     * 设置用户名密码
     *
     * @param username 用户名
     * @param password 密码
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient auth(String username, String password) {
        this.username = username;
        this.password = password;
        return this;
    }

    /**
     * 设置broker地址
     *
     * @param broker broker地址
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient broker(String broker) {
        this.broker = broker;
        return this;
    }

    /**
     * 设置clientId
     *
     * @param clientId clientId
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient clientId(String clientId) {
        this.clientId = clientId;
        return this;
    }

    /**
     * 设置cleanSession
     *
     * @param cleanSession 如果false，保留session。默认true
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient cleanSession(boolean cleanSession) {
        this.cleanSession = cleanSession;
        return this;
    }

    /**
     * 设置是否自动重连
     *
     * @param automaticReconnect true，自动重连。默认false
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient automaticReconnect(boolean automaticReconnect) {
        this.automaticReconnect = automaticReconnect;
        return this;
    }

    /**
     * 设置默认的qos
     *
     * @param qos qos，0,1,2
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient defaultQos(int qos) {
        MqttMessage.validateQos(qos);
        this.defaultQos = qos;
        return this;
    }

    /**
     * 订阅主题，使用默认的qos
     *
     * @param topic 主题
     * @return 返回IotMqttClient对象
     * @see #defaultQos 设置全局qos等级
     */
    public IotMqttClient subscribe(String topic) {
        this.subscribe(topic, defaultQos);
        return this;
    }

    /**
     * 订阅主题
     *
     * @param topic 主题
     * @param qos   qos
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient subscribe(String topic, int qos) {
        this.subTopics.add(new SubTopic(topic, qos));
        return this;
    }

    /**
     * 设置消息回调
     *
     * @param callback 回调
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient callback(BaseMqttCallback callback) {
        this.callback = callback;
        return this;
    }

    /**
     * 设置持久化
     *
     * @param persistence 持久化对象
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient persistence(MqttClientPersistence persistence) {
        this.persistence = persistence;
        return this;
    }

    /**
     * 设置连接配置
     *
     * @param mqttConnectOptions 连接配置
     * @return 返回IotMqttClient对象
     */
    public IotMqttClient mqttConnectOptions(MqttConnectOptions mqttConnectOptions) {
        this.options = mqttConnectOptions;
        return this;
    }

    /**
     * 连接到broker。此方法也可用于重连
     * <p>
     * 执行此方法才会真正创建MqttClient对象并连接
     * </p>
     *
     * @throws MqttException 连接错误抛出异常
     */
    public IotMqttClient connect() throws MqttException {
        this.check();
        if (this.client == null) {
            if (persistence == null) {
                persistence = new MemoryPersistence();
            }
            client = new MqttClient(broker, clientId, persistence);
            // 设置回调
            this.callback.setIotMqttClient(this);
            this.client.setCallback(callback);
            // 建立连接
        }
        this.doConnect();
        return this;
    }

    /**
     * 发布消息，使用默认qos
     *
     * @param topic   主题
     * @param payload 消息内容
     * @throws MqttException MQTT异常
     * @see #defaultQos
     */
    public void publish(String topic, byte[] payload) throws MqttException {
        this.publish(topic, payload, defaultQos);
    }

    /**
     * 发布消息
     *
     * @param topic   主题
     * @param payload 内容
     * @param qos     qos
     * @throws MqttException MQTT异常
     */
    public void publish(String topic, byte[] payload, int qos) throws MqttException {
        this.publish(topic, payload, qos, false);
    }

    /**
     * 发布消息
     *
     * @param topic    主题
     * @param payload  消息体
     * @param qos      qos等级，0、1、2
     * @param retained 是否retain消息
     * @throws MqttException 用于发布消息时遇到的其他错误。例如客户端未连接。
     */
    public void publish(String topic, byte[] payload, int qos, boolean retained) throws MqttException {
        this.client.publish(topic, payload, qos, retained);
    }

    /**
     * 发布消息
     *
     * @param topic       主题
     * @param mqttMessage 消息内容
     * @throws MqttException 用于发布消息时遇到的其他错误。例如客户端未连接。
     */
    public void publish(String topic, MqttMessage mqttMessage) throws MqttException {
        this.client.publish(topic, mqttMessage);
    }

    protected void check() {
        Objects.requireNonNull(broker, "broker can not null");
        Objects.requireNonNull(clientId, "clientId can not null");
    }

    protected void doConnect() throws MqttException {
        if (client == null || !client.isConnected()) {
            if (options == null) {
                options = this.createOptions();
            }
            this.client.connect(options);
        }
    }

    protected MqttConnectOptions createOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        if (username != null && password != null) {
            options.setUserName(username);
            options.setPassword(password.toCharArray());
        }
        options.setCleanSession(cleanSession);
        options.setAutomaticReconnect(automaticReconnect);
        return options;
    }

    /**
     * 是否已连接
     *
     * @return true：已连接
     */
    public boolean isConnected() {
        return client != null && client.isConnected();
    }

    /**
     * 订阅主题
     *
     * @throws MqttException 订阅失败
     */
    public void subscribeTopics() throws MqttException {
        for (SubTopic subTopic : this.subTopics) {
            this.client.subscribe(subTopic.topic, subTopic.qos);
        }
    }

    /**
     * 获取订阅的主题
     *
     * @return 返回订阅的主题
     */
    public List<SubTopic> getSubTopics() {
        return subTopics;
    }

    public MqttClientPersistence getPersistence() {
        return persistence;
    }

    public MqttClient getClient() {
        return client;
    }

    public String getClientId() {
        return clientId;
    }

    public String getBroker() {
        return broker;
    }

    /**
     * 总连接次数，包括重试次数，重试次数=连接次数-1
     *
     * @return 返回次数
     */
    public int getConnectTimes() {
        return connectTimes.get();
    }

    public static abstract class BaseMqttCallback implements MqttCallbackExtended {
        private IotMqttClient iotMqttClient;

        /**
         * 连接成功会进入到这里. <br />
         * 如果设置了<code>IotMqttClient.automaticReconnect(true)</code>，断开连接后会自动进入这里<br />
         * 如果设置<code>IotMqttClient.automaticReconnect(false)</code>，
         * {@link org.eclipse.paho.client.mqttv3.MqttCallback#connectionLost(Throwable)}实现重连功能
         *
         * @param reconnect 是否重连
         * @param serverURI 服务器URI
         */
        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
            int connectTimes = iotMqttClient.connectTimes.incrementAndGet();
            LOG.info("{}MqttClient连接完成，连接次数={}, clientId={}, broker={}, cleanSession={}, automaticReconnect={}",
                    logTitle,
                    connectTimes,
                    iotMqttClient.clientId,
                    iotMqttClient.broker,
                    iotMqttClient.cleanSession,
                    iotMqttClient.automaticReconnect
            );
            //只有订阅有主题，发布不走这段代码
            if (!this.iotMqttClient.getSubTopics().isEmpty()) {
                try {
                    this.iotMqttClient.subscribeTopics();
                    LOG.info("{}订阅主题成功，clientId={}, topics={}", logTitle, iotMqttClient.clientId, iotMqttClient.subTopics);
                } catch (MqttException e) {
                    LOG.error("{}订阅主题失败, reasonCode={}, broker={}, clientId={}, topics={}",
                            logTitle,
                            e.getReasonCode(),
                            this.iotMqttClient.broker, iotMqttClient.clientId, iotMqttClient.subTopics, e);
                }
            }
        }

        public IotMqttClient getIotMqttClient() {
            return iotMqttClient;
        }

        public void setIotMqttClient(IotMqttClient iotMqttClient) {
            this.iotMqttClient = iotMqttClient;
        }
    }

    protected static class SubTopic {
        private final String topic;
        private final int qos;

        public SubTopic(String topic, int qos) {
            this.topic = topic;
            this.qos = qos;
        }

        @Override
        public String toString() {
            return "SubTopic{" +
                    "topic='" + topic + '\'' +
                    ", qos=" + qos +
                    '}';
        }
    }
}